package com.spark.gaea.mq.rabbit.config;

import com.google.common.base.Preconditions;
import com.spark.gaea.config.service.easyzk.ConfigNode;
import com.spark.gaea.crypto.Decrypter;
import com.spark.gaea.utils.StringUtil;
import com.spark.gaea.utils.exception.GaeaRuntimeException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.AmqpAdmin;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.support.converter.ClassMapper;
import org.springframework.amqp.support.converter.DefaultClassMapper;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.env.Environment;

import java.util.HashMap;

@Configuration
public abstract class AbstractRabbitConfig {
	private static final Logger LOG;
	public final String ADRESS = "CONFIG_ZK_ADDRESSES";
	public final String NODE = "CONFIG_ZK_NODE";
	public final String MISID = "misid";
	public final String EVN = "EVN";
	@Autowired
	private Environment env;
	@Autowired
	@Qualifier("applicationConfigNode")
	private ConfigNode applicationConfigNode;
	private static final String DEFAULT_CHANNEL_CACHE_SIZE = "25";

	protected abstract void configureRabbitTemplate(final RabbitTemplate p0);

	@Bean
	public ConnectionFactory connectionFactory() {
		final ConfigNode configNode = this.applicationConfigNode;
		System.out.println(StringUtil.getProperty(configNode, this.env, "rabbit.addresses"));
		final String addresses = Preconditions
				.checkNotNull(StringUtil.getProperty(configNode, this.env, "rabbit.addresses"),
						"rabbit.addresses is null");
		final String username = Preconditions
				.checkNotNull(StringUtil.getProperty(configNode, this.env, "rabbit.username"),
						"rabbit.username is null");
		final boolean isEncrypt = Boolean
				.parseBoolean(StringUtil.getProperty(configNode, this.env, "rabbit.encrypt", "false"));
		String password = Preconditions.checkNotNull(StringUtil.getProperty(configNode, this.env, "rabbit.password"),
				"rabbit.password is null");
		final String channelCacheSize = StringUtil.getProperty(configNode, this.env, "rabbit.channelCacheSize", "25");
		final CachingConnectionFactory connectionFactory = new CachingConnectionFactory();
		connectionFactory.setAddresses(addresses);
		connectionFactory.setUsername(username);
		if (isEncrypt) {
			try {
				password = Decrypter.decrypter(password);
			} catch (Exception e) {
				throw new GaeaRuntimeException("RabbitMQ\u5bc6\u7801\u89e3\u5bc6\u5931\u8d25", e);
			}
		}
		connectionFactory.setPassword(password);
		connectionFactory.setChannelCacheSize(Integer.parseInt(channelCacheSize));
		return connectionFactory;
	}

	@Bean
	public AmqpAdmin amqpAdmin() {
		return new RabbitAdmin(this.connectionFactory());
	}

	@Bean
	public RabbitTemplate rabbitTemplate() {
		final RabbitTemplate template = new RabbitTemplate(this.connectionFactory());
		template.setMessageConverter(this.jackson2JsonMessageConverter());
		this.configureRabbitTemplate(template);
		return template;
	}

	@Bean
	public ClassMapper classMapper() {
		final DefaultClassMapper classMapper = new DefaultClassMapper();
		classMapper.setDefaultType(HashMap.class);
		classMapper.setDefaultHashtableClass(HashMap.class);
		return classMapper;
	}

	@Bean
	public Jackson2JsonMessageConverter jackson2JsonMessageConverter() {
		final Jackson2JsonMessageConverter converter = new Jackson2JsonMessageConverter();
		converter.setClassMapper(this.classMapper());
		return converter;
	}

	static {
		LOG = LoggerFactory.getLogger(AbstractRabbitConfig.class);
	}
}
